package com.cxz.kotlin.baselibs.rx.bus;


import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.processors.FlowableProcessor;
import io.reactivex.rxjava3.processors.PublishProcessor;

/**
 * Created by Czg on 2018/9/7
 */
public class RxBus {

    private final FlowableProcessor<Object> bus;

    private RxBus() {
        bus = PublishProcessor.create().toSerialized();
    }

    public static RxBus get() {
        return Holder.BUS;
    }

    public void send(Object obj) {
        bus.onNext(obj);
    }
    /**
     * 根据传递的 eventType 类型返回特定事件类型的被观察者
     */
    public <T> Flowable<T> toFlowable(Class<T> tClass) {
        return bus.ofType(tClass);
    }

    public Flowable<Object> toFlowable() {
        return bus;
    }

    public boolean hasSubscribers() {
        return bus.hasSubscribers();
    }

    public void unSubscribe(Disposable disposable) {
        if (disposable != null) {
            disposable.dispose();
            disposable = null;
        }
    }

    private static class Holder {
        private static final RxBus BUS = new RxBus();
    }

}
